package chapter4.eg10;

import chapter4.AbstractProducer;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Test;

public class ProducerTest extends AbstractProducer {
  @Test
  public void testProducer() throws ExecutionException, InterruptedException {
    properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "chapter4.eg10.AuditPartitioner");
    String topic = "test-topic";
    Producer<String, String> producer = new KafkaProducer<>(properties);
    ProducerRecord<String, String> nonKeyRecord = new ProducerRecord<>(topic, "non-key topic");
    ProducerRecord<String, String> auditRecord = new ProducerRecord<>(topic, "audit", "audit record");
    ProducerRecord<String, String> nonAuditRecord = new ProducerRecord<>(topic, "other", "non-audit record");
    RecordMetadata recordMetadata = producer.send(nonKeyRecord).get();
    System.out.println(recordMetadata.partition());
    recordMetadata = producer.send(auditRecord).get();
    System.out.println(recordMetadata.partition());
    recordMetadata = producer.send(nonAuditRecord).get();
    System.out.println(recordMetadata.partition());
    producer.close();
  }

}
